/**
 * log user info to table hbase
 * Main task log data 
 */

package olap4cloud.hadop.statistic;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletionService;
import java.util.concurrent.ExecutorService;

import olap4cloud.hbase.TrackingTable;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.client.HTablePool;
import org.apache.hadoop.hbase.client.Increment;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.io.SequenceFile.Reader;
import org.apache.hadoop.util.ReflectionUtils;

/**
 * @author  LuyenThanhDat
 */
public class IncreDataToHbase implements Callable {

	private static final int MAXROWINBATCH = 500;
	private static final int MAX__NO__OF__BATCH = 500;

	/**
	 * @uml.property  name="objectCompletionService"
	 */
	private CompletionService<Object> objectCompletionService;
	/**
	 * @uml.property  name="eservice"
	 */
	private ExecutorService eservice;
	/**
	 * @uml.property  name="pool"
	 * @uml.associationEnd  multiplicity="(1 1)"
	 */
	private HTablePool pool;
	/**
	 * @uml.property  name="filePut"
	 * @uml.associationEnd  multiplicity="(1 1)"
	 */
	private Path filePut;

	public static int TOTAL_LINE_INCRE = 0;

	public IncreDataToHbase(HTablePool pool, Path filePut) {
		this.pool = pool;
		this.filePut = filePut;
	}

	/**
	 * @return
	 * @uml.property  name="objectCompletionService"
	 */
	public CompletionService<Object> getObjectCompletionService() {
		return objectCompletionService;
	}

	/**
	 * @param objectCompletionService
	 * @uml.property  name="objectCompletionService"
	 */
	public void setObjectCompletionService(
			CompletionService<Object> objectCompletionService) {
		this.objectCompletionService = objectCompletionService;
	}

	/**
	 * @return
	 * @uml.property  name="eservice"
	 */
	public ExecutorService getEservice() {
		return eservice;
	}

	/**
	 * @param eservice
	 * @uml.property  name="eservice"
	 */
	public void setEservice(ExecutorService eservice) {
		this.eservice = eservice;
	}

	@Override
	public Object call() throws Exception {
		// TODO Auto-generated method stub
		int countNoOfBatch = 0;
		List<Increment> listIncre = new ArrayList<Increment>();
		Configuration conf = new Configuration();
		FileSystem fs = FileSystem.get(new Configuration());
		SequenceFile.Reader reader = new Reader(fs, this.filePut, conf);

		Model key = (Model) ReflectionUtils.newInstance(
				reader.getKeyClass(), conf);
		LongWritable value = (LongWritable) ReflectionUtils
				.newInstance(reader.getValueClass(), conf);
		
		int indext = 0;
		int count = 0;
		
		while (reader.next(key, value)) {
			indext++;
			count++;
	
			System.out.println(key +" "+ value);
			
			Increment incre = new Increment(key.getKey());
			incre.addColumn(key.getColumF(), key.getColumQ(), value.get());
			listIncre.add(incre);
			
			if (count >= MAXROWINBATCH) {
				TaskIncreDataInToDB taskPut = new TaskIncreDataInToDB(
						pool.getTable(StatisticsTable.getTableName()), listIncre);
				this.objectCompletionService.submit(taskPut);
				count = 0;
				
				countNoOfBatch++;
				if (countNoOfBatch >= MAX__NO__OF__BATCH) {
					for (int i = 0; i < countNoOfBatch; i++) {
						Object o = objectCompletionService.take().get();
						countNoOfBatch = 0;
					}
				}
				
				listIncre.clear();
			}
		}
		reader.close();
		if (count > 0) {
			TaskIncreDataInToDB taskPut = new TaskIncreDataInToDB(
					pool.getTable(StatisticsTable.getTableName()), listIncre);
			this.objectCompletionService.submit(taskPut);
			listIncre.clear();
			count = 0;
		}

		for (int i = 0; i < countNoOfBatch; i++) {
			Object o = objectCompletionService.take().get();
		}

		System.out.println("Total line is : " + indext);
		return true;
	}

}
